package com.atguigu.flink.day09;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2024/8/19
 * 该案例演示了算子状态-列表状态
 * 需求：在process算子中的每个并行度上计算数据的个数
 */
public class Flink06_OpeState_ListState_1 {
    public static void main(String[] args) throws Exception {
        //TODO 1.环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(2);
        //TODO 2.从指定的网络端口读取数据
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);
        //TODO 3.在process算子中的每个并行度上计算数据的个数
        socketDS.process(
                new ProcessFunction<String, String>() {
                    //用于存储每个并行子任务计算元素的个数
                    Integer count = 0;
                    @Override
                    public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
                        out.collect("当前子任务"+getRuntimeContext().getIndexOfThisSubtask()+"计算的元素的个数为" + ++count);
                    }
                }
        ).print();


        env.execute();

    }
}
